/*********************************************************
*Copyright (c) 2009 Rob Groves. All Rights in Pw 
*@des: 异步TCP Sever
*@name: Server.cpp
*@author: zhognsx
*@time: 2015-02-27
*********************************************************/

#include "Server.h"
#include <iostream>

io_service_pool::io_service_pool(std::size_t pool_size): next_io_service_(0)
{
    for (std::size_t i = 0; i < pool_size; ++ i)
    {
        io_service_sptr io_service(new boost::asio::io_service);
		work_sptr work(new boost::asio::io_service::work(*io_service));
		io_services_.push_back(io_service);
		work_.push_back(work);
   	}
}

void io_service_pool::start()
{
    for (std::size_t i = 0; i < io_services_.size(); ++ i)
    {
        boost::shared_ptr<boost::thread> thread(new boost::thread(
            boost::bind(&boost::asio::io_service::run, io_services_[i])));
        threads_.push_back(thread);
    }
}

void io_service_pool::join()
{
    for (std::size_t i = 0; i < threads_.size(); ++ i)
    {
        threads_[i]->join();
    }
}

void io_service_pool::stop()
{
    for (std::size_t i = 0; i < io_services_.size(); ++ i)
    {
        io_services_[i]->stop();
    }
}

boost::asio::io_service& io_service_pool::get_io_service()
{
    boost::mutex::scoped_lock lock(mtx);
    boost::asio::io_service& io_service = *io_services_[next_io_service_];
    ++ next_io_service_;
    if (next_io_service_ == io_services_.size())
    {
        next_io_service_ = 0;
    }
    return io_service;
}



////////////////////////////////////////////////////////////////////////
session::session(boost::asio::io_service& ios, CallBack_SF _func):
	io_service_(ios), socket_(ios), func(_func), timer_(ios)
{
	read_buf_ = boost::shared_ptr<message>(new message());
}

session::~session()
{

}

boost::asio::ip::tcp::socket& session::socket()
{
	return socket_;
}

void session::start_read()
{
	//连接成功, 调用读取
	read_buf_->reset();
	//读取数据报头 
	boost::asio::async_read(socket_,  
		boost::asio::buffer(read_buf_->header(), message::header_length), 
		boost::bind(&session::handle_read_header, this, _1, _2));  
	//
	timer_.expires_from_now(boost::posix_time::seconds(10));
	//timer_.expires_at(timer_.expires_at() + boost::posix_time::seconds(10)); 
	timer_.async_wait(boost::bind(&session::time_out, this, _1));
}

void session::write(boost::shared_ptr<message> msg)  
{  
	//将消息主动投递给io_service  
	io_service_.post(boost::bind(&session::do_write, this, msg)); 
}  

void session::close()  
{
	//读数据失败时调用    
	//这个close函数是客户端要主动终止时调用  do_close函数是从服务器端  
	io_service_.post(boost::bind(&session::do_close, this)); 
}  

//设置错误回调
void session::SetErrorCallBack(CallBack_SE _error)
{
	error = _error;
}

void session::handle_read_header(const boost::system::error_code& ec, int bytes_)  
{  
	//分别处理数据报头和数据部分  
	if (!ec) {  
		//处理数据头,获取body数据长度
		if (bytes_ == message::header_length){
			//msgHead_Main* head = read_bufs_->decode_header();
			int blen = read_buf_->body_length();
			//异步读取指定长度hlen的body数据
			boost::asio::async_read(socket_, boost::asio::buffer(read_buf_->body(), blen),	
				boost::bind(&session::handle_read_msg, this, _1, _2));

			//
			timer_.expires_at(timer_.expires_at() + boost::posix_time::seconds(10));  
			timer_.async_wait(boost::bind(&session::time_out, this, _1));
		}
		else{
			start_read();
		}
	}  
	else  {
		errstr = ec.message();
		boost::shared_ptr<session> pSession = this->shared_from_this();
		if(error){
			error(pSession);
		}
	}  
}  

void session::handle_read_msg(const boost::system::error_code& ec, int bytes_)  
{  
	boost::shared_ptr<session> pSession = this->shared_from_this();
	if (!ec){
		if (bytes_ == read_buf_->body_length()){
			//生成处理消息
			boost::shared_ptr<message> msg(new message());
			msg->put_data(read_buf_->data(), read_buf_->length());
			//调用处理函数
			func(pSession, msg);
		}
		//继续
		start_read();
	}  
	else{  
		errstr = ec.message();
		if(error){
			error(pSession);
		}
	}  
}  

void session::do_write(boost::shared_ptr<message> msg)  
{
	bool write_in_progress = !write_bufs_.empty(); //空的话变量为false  
	write_bufs_.push_back(msg); //把要写的数据push至写队列  
	if (!write_in_progress)//队列初始为空 push一个msg后就有一个元素了  
	{  
		boost::shared_ptr<message> buf = write_bufs_.front();
		// 
		boost::asio::async_write(socket_, boost::asio::buffer(buf->data(), buf->length()), 
			boost::bind(&session::handle_write, this, _1));  		
	}  
}  

//第一个消息单独处理，剩下的才更好操作  
void session::handle_write(const boost::system::error_code& ec)
{  
	if (!ec){  
		write_bufs_.pop_front();//刚才处理完一个数据 所以要pop一个  
		if (!write_bufs_.empty())    
		{          		
			//循环处理剩余的消息 
			boost::shared_ptr<message> buf = write_bufs_.front();
			// 
			boost::asio::async_write(socket_, boost::asio::buffer(buf->data(), buf->length()),
				boost::bind(&session::handle_write, this, _1));  	 		
		}  
	}  
	else{ 
		boost::shared_ptr<session> pSession = this->shared_from_this();
		errstr = ec.message();
		if(error){
			error(pSession);
		}
	}  
}  

void session::do_close()  
{  
	socket_.shutdown(boost::asio::socket_base::shutdown_both);
	//sta = CLOSE;
	//boost::shared_ptr<session> pSession = this->shared_from_this();
	//ClientMgr::GetClietMgr()->pop(pClient);
	//socket_.close(); 
} 

void session::time_out(const boost::system::error_code& ec)
{
	if (!ec){
		errstr = "The time out.";
		if(error){
			boost::shared_ptr<session> pSession = this->shared_from_this();
			error(pSession);
		}

		//do_close();
	}
}

void session::SetSessionInfo(string ip, short port)
{
	sessionIp = ip;
	sessionPort = port;
}

/////////////////////////////////////////////////////////////
server::server(short port, CallBack_SF _func): func(_func), io_service_pool_(1),
	acceptor_(io_service_pool_.get_io_service(), 
	boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port))
{
	// boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::tcp::v4(), port);
	// acceptor_.open(endpoint.protocol());
	// acceptor_.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
	// acceptor_.bind(endpoint);
	// acceptor_.listen();

	//
	boost::shared_ptr<session> new_session(
			new session(io_service_pool_.get_io_service(), func));
	//
	sesion_vec.push_back(new_session);
	//ClientMgr::GetClietMgr()->push(new_session);

	acceptor_.async_accept(new_session->socket(),
            boost::bind(&server::handle_accept, this, new_session,
            boost::asio::placeholders::error));
}

void server::run()
{
	io_service_pool_.start();
    //io_service_pool_.join();
}

void server::handle_accept(boost::shared_ptr<session>& new_session, 
			const boost::system::error_code& error)
{
	if (!error){
		string ip = new_session->socket().remote_endpoint().address().to_string();
		short port = new_session->socket().remote_endpoint().port();
		new_session->SetSessionInfo(ip, port);
		//响应客户请求，并等待下个连接
		new_session->start_read();
		//
		boost::shared_ptr<session> _session(
			new session(io_service_pool_.get_io_service(), func));
		//
		sesion_vec.push_back(_session);
		//ClientMgr::GetClietMgr()->push(_session);
	
		acceptor_.async_accept(_session->socket(),
			boost::bind(&server::handle_accept, this, _session,
			boost::asio::placeholders::error));
	}
	else{
		return ;
	}
}

